package com.navinfo.platform.trip.analysis.flink.watermark;

import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * 重写{@link org.apache.flink.api.common.eventtime.BoundedOutOfOrdernessWatermarks}
 * 加入事件跳变的最大范围处理，防止事件时间跳变过大
 */
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {
	/** The maximum timestamp encountered so far */
	private long maxTimestamp;

	/** The maximum out-of-orderness that this watermark generator assumes.*/
	private final long outOfOrdernessMillis;

	/**
	 * 事件时间跳变的边界范围，单位ms
	 */
	private final long changeBoundaryMillis;

	/**
	 * Creates a new watermark generator with the given out-of-orderness bound.
	 *
	 * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
	 * @param maxChangeBoundary The bound for the change of the event timestamps.
	 */
	public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness, Duration maxChangeBoundary) {
		checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
		checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
		checkNotNull(maxChangeBoundary, "maxChangeBoundary");
		checkArgument(!maxChangeBoundary.isNegative(), "maxChangeBoundary cannot be negative");

		this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
		this.changeBoundaryMillis = maxChangeBoundary.toMillis();

		// start so that our lowest watermark would be Long.MIN_VALUE.
		this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
	}

	// ------------------------------------------------------------------------

	@Override
	public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
		if((maxTimestamp==0) || (eventTimestamp-maxTimestamp < changeBoundaryMillis)) {
			maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
		}
	}

	@Override
	public void onPeriodicEmit(WatermarkOutput output) {
		output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
	}
}
